package com.pw.study.flink.state;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class KeyStateValue {

    public static void main(String[] args) {
        stateList();
    }

    private static void stateList() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        //输入数据
        SingleOutputStreamOperator<WaterSensor> stream = env.socketTextStream("hadoop111", 9999).map(value -> {
            String[] line = value.split(",");
            return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));
        });

        //处理数据
        stream.keyBy(WaterSensor::getId)
                .flatMap(new RichFlatMapFunction<WaterSensor, String>() {
                             private ValueState<Integer> lastVcState;

                             @Override
                             public void open(Configuration parameters) throws Exception {
                                 lastVcState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("lastVcState", Integer.class));
                             }

                             @Override
                             public void flatMap(WaterSensor value, Collector<String> out) throws Exception {
                                 System.out.println("last: "+lastVcState.value());
                                 if(lastVcState.value()!=null){
                                     Integer lastVc = lastVcState.value();
                                     if(lastVc >10 &&value.getVc()>10){
                                         System.out.println("last: "+lastVc+" current: "+value.getVc());
                                        out.collect(value.getId()+"连续两次水位超过10.。。发出红色告警++++++++=");
                                     }
                                 }
                                 lastVcState.update(value.getVc());
                             }
                         }
                ).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
